use std::any::Any;
use std::collections::HashMap;
use std::fmt::{Debug, Display};
use std::sync::{Arc};
use std::sync::atomic::{AtomicU32, Ordering};
use std::time::{SystemTime, UNIX_EPOCH};
use async_trait::async_trait;
use bincode::{Decode, Encode};
use chrono::{DateTime, Local};
use log::{error};
use serde::de::DeserializeOwned;
use serde::Serialize;
use tokio::sync::mpsc::{UnboundedSender};
use crate::codec::{RpcCodecGather};
use crate::codec::binary_codec::{RpcBinCodeCodec};
use crate::codec::json_codec::{RpcJsonCodec};
use crate::connection_manager::event::{ConnectionEvent};
use crate::dto::{RpcData, RpcHeader};
use crate::error::{RpcError};
use crate::sender::RpcSender;
use crate::session::RpcSession;
use crate::util::{get_current_rpc_codec, get_current_rpc_header_codec};

pub mod error;
pub mod codec;
pub mod router;

pub mod dto;
pub mod connection_manager;
pub mod session;
pub mod station;
pub mod sender;
pub mod bootstrap;
mod util;
mod io_util;
mod test;

pub fn spawn<F, E>(future: F)
where
    F: Future<Output = Result<(), E>> + Send + 'static,
    E: Display,
{
    tokio::task::spawn(async {
        if let Err(error) = future.await {
            error!("{}", error);
        }
    });
}

pub fn send<T>(sender: &UnboundedSender<T>, message: T) -> bool {
    match sender.send(message) {
        Ok(_) => true,
        Err(error) => {
            error!("send message error: {}", error);
            false
        },
    }
}

pub type RpcResult<T> = Result<T, RpcError>;

pub trait RpcEvent: Send + Sync {
    fn consume(&self, context: RpcContext)-> RpcResult<()>;
}

pub trait RpcFilter: Send + Sync {
    fn before(&self, _: &mut RpcContext, _: &mut dyn Any) -> RpcResult<bool> {
        Ok(true)
    }

    fn after(&self, rpc_channel: &mut RpcContext, data: &mut dyn Any) -> Result<(), RpcError>;
}

pub trait LocalSeqCreator: Send + Sync {
    fn create_seq(&self) -> String;

    fn new(prefix: String) -> Self where Self: Sized;

    fn build(&self) -> Self where Self: Sized;

}

// pub trait LocalSeqCreatorFactory: Clone {
//     fn create(&self)-> Box<dyn LocalSeqCreator>;
// }

#[derive(Clone)]
pub struct RpcContext{
    session: RpcSession,
    sender: RpcSender,
}

unsafe impl Send for RpcContext {}
unsafe impl Sync for RpcContext {}

impl RpcContext {

    pub fn new(sender: RpcSender) -> Self {
        let now: DateTime<Local> = Local::now();
        let prefix = format!("{}", now.format("%Y%m%d%H%M%S"));
        Self {
            session: RpcSession::new(Arc::new(Box::new(DefaultLocalSeqCreator::new(prefix))), create_default_rpc_codecs()),
            sender,
        }
    }

    pub fn create(local_seq_creator: Arc<Box<dyn LocalSeqCreator>>, rpc_codec_gather: RpcCodecGather, sender: RpcSender) -> Self {
        Self {
            session: RpcSession::new(local_seq_creator, rpc_codec_gather),
            sender,
        }
    }

    pub fn session(&self)-> RpcSession {
        self.session.clone()
    }

    pub fn send_event(&mut self, event: Box<dyn RpcEvent>) -> RpcResult<()> {
        self.sender.send_event(event)
    }

    pub fn send_data(&mut self, event: Box<dyn ConnectionEvent>) -> RpcResult<()> {
        self.sender.send_data(event)
    }

    pub fn get_current_rpc_codec_gather(&self) -> RpcCodecGather {
        self.session().get_rpc_codec_gather()
    }

    pub fn set_current_codec_type(&mut self, codec_type: Option<String>) -> RpcResult<()> {
        self.session().set_current_codec_type(codec_type)
    }

    pub fn get_local_seq_creator(&self) -> Arc<Box<dyn LocalSeqCreator>> {
        self.session().local_seq_creator()
    }

    pub fn get_current_codec_type(&self) -> Option<String> {
        self.session().get_current_codec_type()
    }

    pub fn sender(&self) -> RpcSender {
        self.sender.clone()
    }
}

pub fn encode_rpc_data<T: Decode<()> + Encode + DeserializeOwned + Serialize + 'static>(rpc_context: &RpcContext, data: &RpcData<T>) -> RpcResult<Vec<u8>> {
    let any : Box<&dyn Any> = Box::new(&data.data);
    if any.is::<Vec<u8>>() {
        let value = any.downcast_ref::<Vec<u8>>().unwrap();
        let codec = get_current_rpc_header_codec(rpc_context)?;
        let bytes = codec.encode_data(&data.header, value)?;
        return Ok(bytes)
    }
    let codec = get_current_rpc_codec::<T>(rpc_context)?;
    let bytes = codec.encode_rpc_data(data)?;

    Ok(bytes)
}

pub fn encode<T: Decode<()> + Encode + DeserializeOwned + Serialize + 'static>(rpc_context: &RpcContext, data: &T)-> RpcResult<Vec<u8>> {
    let codec = get_current_rpc_codec::<T>(rpc_context)?;
    let bytes = codec.encode(data)?;
    Ok(bytes)
}

pub fn decode<T: Decode<()> + Encode + DeserializeOwned + Serialize + 'static>(rpc_context: &RpcContext, data: Vec<u8>)-> RpcResult<T>{
    let codec = get_current_rpc_codec::<T>(rpc_context)?;
    codec.decode(data)
}

pub fn decode_rpc_header(rpc_context: &RpcContext, data: &Vec<u8>) -> RpcResult<(RpcHeader, Vec<u8>)> {
    let decode_header = get_current_rpc_header_codec(rpc_context)?;
    decode_header.decode_header(data)
}


#[async_trait]
pub trait RpcChannel: Send + Sync {

    async fn write(&mut self, local_seq: String, data: Vec<u8>, timeout: u32) -> RpcResult<Vec<u8>>;
}

#[derive(Clone, Hash, Eq, PartialEq, Debug, Copy)]
pub struct ChannelId {
    id: u32,
}

unsafe impl Send for ChannelId {}
unsafe impl Sync for ChannelId {}

impl ChannelId {
    pub fn new(id: u32) -> Self {
        ChannelId { id }
    }
}

pub fn create_default_rpc_codecs() -> RpcCodecGather {
    let mut map : HashMap<String, Box<dyn Any>> = HashMap::new();
    map.insert("json".to_string(), Box::new(RpcJsonCodec::default()));
    map.insert("bin-code".to_string(), Box::new(RpcBinCodeCodec::default()));
    RpcCodecGather::new(map)
}

#[derive(Debug, Clone)]
pub struct DefaultLocalSeqCreator {
    prefix: String,
    inc_seq: Arc<AtomicU32>,
}

unsafe impl Send for DefaultLocalSeqCreator {}
unsafe impl Sync for DefaultLocalSeqCreator {}

// #[derive(Debug, Clone)]
// pub struct DefaultLocalSeqCreatorFactory {
//     instant: DateTime<Local>,
// }

impl DefaultLocalSeqCreator {
    pub fn new(prefix: String) -> DefaultLocalSeqCreator {
        DefaultLocalSeqCreator {
            prefix,
            inc_seq: Arc::new(AtomicU32::new(0)),
        }
    }

    fn create_seq_0(&self) -> String {
        let seq = self.inc_seq.fetch_add(1, Ordering::Relaxed);
        format!("{}-{}", self.prefix, seq)
    }
}

impl LocalSeqCreator for DefaultLocalSeqCreator {
    fn create_seq(&self) -> String {
        self.create_seq_0()
    }

    fn new(prefix: String) -> Self where Self: Sized {
        DefaultLocalSeqCreator::new(prefix)
    }

    fn build(&self) -> Self where Self: Sized {
        DefaultLocalSeqCreator::new(self.prefix.clone())
    }
}

// impl DefaultLocalSeqCreatorFactory {
//     pub fn default() -> DefaultLocalSeqCreatorFactory {
//         DefaultLocalSeqCreatorFactory { instant: Local::now() }
//     }
//
//     pub fn new(instant: DateTime<Local>) -> DefaultLocalSeqCreatorFactory {
//         DefaultLocalSeqCreatorFactory { instant }
//     }
// }

// impl LocalSeqCreatorFactory for DefaultLocalSeqCreatorFactory {
//     fn create(&self) -> Box<dyn LocalSeqCreator> {
//         let prefix = format!("{}", self.instant.format("%Y%m%d%H%M%S"));
//         Box::new(DefaultLocalSeqCreator::new(prefix))
//     }
// }


pub fn time_from_millis(millis: u64) -> SystemTime {
    UNIX_EPOCH + std::time::Duration::from_millis(millis)
}

pub fn time_to_millis(now: SystemTime)-> u64 {
    if let Ok(duration) = now.duration_since(UNIX_EPOCH) {
        let milliseconds = duration.as_secs() * 1000 + u64::from(duration.subsec_millis());
        milliseconds
    } else {
        0u64
    }
}